package offset_gaobo

import kafka.common.TopicAndPartition
import kafka.message.MessageAndMetadata
import kafka.serializer.StringDecoder
import offset_gaobo.util.OffSetUtils
import org.apache.spark.SparkConf
import org.apache.spark.streaming.kafka.{HasOffsetRanges, KafkaUtils}
import org.apache.spark.streaming.{Seconds, StreamingContext}

import scala.collection.immutable

// spark作为消费者接收数据
object SparkConsumer {


  def main(args: Array[String]): Unit = {

    // 参数效验
    if (args.length != 4) {
      println("参数有误")
      // 终止JVM运行 返回状态码 1
      System.exit(1)
    }

    // 配置sparkStreamingContext
    val sparkConf = new SparkConf().setAppName("Goobo").setMaster("local[6]").set("spark.testing.memory", "471859200")
    // 每30秒生成一个RDD
    val ssc = new StreamingContext(sparkConf, Seconds(30))
    //ssc.sparkContext.setLogLevel("ERROR")
    // 用一个元组接收 ages参数
    val Array(zkCluster, consumerGroupId, topics, numThreads) = args


    //  Map   broker 元数据  和 offset配置
    val map = Map[String, String](
      "metadata.broker.list" -> "zocoo:9090,zocoo:9091,zocoo:9092",
      // auto 自动 reset 重置    smallest 最小的
      // 如果自定义的offset出现异常 就会调用当前最小的
      "auto.offset.reset" -> "smallest"
    )


    // topic参数  在这个案例只有一个 offset_topic参数
    val offsettopic = topics.split(",").toSet

    /*
      从zookeeper获取ofset信息
          {"topic":"offsetTopic","partition":"0","from":"100","until":"200"}
          {"topic":"offsetTopic","partition":"1","from":"12","until":"300"}
          {"topic":"offsetTopic","partition":"2","from":"200","until":"400"}



     */

    //todo 生成map工具类方法编写
    val offsetMap1:immutable.Map[TopicAndPartition, Long] =OffSetUtils.getOffsetMap.toMap

    // 用一个函数 获取 message
    val messages = if (offsetMap1.size == 0) {
      println( "!!!!!!!!!!!    if里 ______  ")
      KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](ssc, map, offsettopic)
    } else {

      // 函数编程
      val messageHandler = (mm: MessageAndMetadata[String, String]) => {

   /*     println("所在分区:" + mm.partition)
        println("offset值:" + mm.offset)
        println("数据:" + mm.message())*/

        // 获取分区存储offset值文件的版本号

        //var version = ZookeeperUtils.getVersion("/p" + String.valueOf(mm.partition))

        // 更新各个分区的 offset值
        //ZookeeperUtils.updataOffset("/p" + String.valueOf(mm.partition),mm.offset+1, version)

        //return
        (  mm.key(), mm.message(), mm.offset ,mm.partition )
      }

      KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder, (String, String,Long,Int)](ssc, map, offsetMap1, messageHandler)

    }


    val unit = messages.foreachRDD(
      rdd => {
        rdd.foreachPartition(
          p => {
            p.foreach(
              line => {
                println("每行数据 : " + line.toString)
              }
            )
          }
        )

        val offsetRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges
        //创建工具类 遍历offsetRanges  并保存offset到zookeeper
        OffSetUtils.saveOffset(offsetRanges)
      }
    )







    /*
        // 提取topic
        val topicMap = topics.split(",").map( line => (line,numThreads.toInt)).toMap
        // 获取消息  用kafka工具类 获得一个DStream
        val messages = KafkaUtils.createStream(ssc,zkCluster,consumerGroupId,topicMap)
        messages.map(
          line =>
            line.toString()
        ).print()*/



    ssc.start()
    ssc.awaitTermination()

  }


}
